"""
Case Type   : 防篡改
Case Name   : 账本数据库基础语法适配(DDL/DML)--enable_copy_server_files打开，系统管理员使用copyto命令导出防篡改用户表，初始用户使用copy to导出历史表和全局区块表
Create At   : 2022-02-28
Owner       : lwx1080719
Description :
    1.查询参数enableSeparationOfDuty和enable_copy_server_files
    2.修改enable_copy_server_files为on
    3.创建不同用户
    4.系统管理员创建schema并指定防篡改选项
    5.普通用户登录数据库在防篡改schema下创建防篡改用户表并插入数据
    6.系统管理员登录数据库使用copyto命令导出防篡改用户表
    7.初始用户使用copy to导出历史表和全局区块表
    8.查询导出的文件是否存在
    9.普通用户登录数据库对防篡改表进行delete
    10.清理环境
Expect      :
    1.显示默认值off
    2.成功
    3.成功
    4.成功
    5.成功
    6.成功
    7.成功
    8.成功
    9.成功
    10.成功
History     :
"""




import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Constant import Constant
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Logger import Logger


class ModifyCase(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f'-----{os.path.basename(__file__)} start-----')
        self.dbuserNode = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.Constant = Constant()
        self.Common = Common()
        self.schema_name = 's_security_tamper_proof_0051'
        self.table_name = 't_security_tamper_proof_0051'
        self.user_path = os.path.join(macro.DB_INSTANCE_PATH, 'user.bat')
        self.hist_path = os.path.join(macro.DB_INSTANCE_PATH, 'blockchain.hist')
        self.glob_path = os.path.join(macro.DB_INSTANCE_PATH, 'globaltable')
        self.user_name = 'u_security_tamper_proof'
        self.user_privileges = {f'{self.user_name}_sys': 'sysadmin',
                                f'{self.user_name}_noruser': ''}
        self.default_value1 = self.Common.show_param('enableSeparationOfDuty')
        self.default_value = self.Common.show_param('enable_copy_server_files')

    def test_security(self):
        text = '----step1:查询参数值  expect:默认值off----'
        self.logger.info(text)
        self.assertEqual("off", self.default_value1, "执行失败:" + text)
        self.assertEqual("off", self.default_value, "执行失败:" + text)

        text = '----step2:修改enable_copy_server_files为on ' \
               'expect:成功----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
                            alter system set enable_copy_server_files to on;
                            select pg_sleep(2);
                            show enable_copy_server_files;''')
        self.logger.info(sql_cmd)
        self.assertEqual("on", sql_cmd.split("\n")[-2].strip(),
                         "执行失败:" + text)

        text = '----step3:创建不同用户 expect:成功----'
        self.logger.info(text)
        for user, privilege in self.user_privileges.items():
            create_cmd = f"drop user if exists {user} cascade; " \
                f"create user {user} {privilege} " \
                f"password '{macro.PASSWD_REPLACE}' ; "
            msg = self.primary_sh.execut_db_sql(create_cmd)
            self.logger.info(msg)
            self.assertIn(self.Constant.DROP_ROLE_SUCCESS_MSG and
                          self.Constant.CREATE_ROLE_SUCCESS_MSG, msg,
                          '执行失败:' + text)

        text = '----step4:系统管理员创建schema并指定防篡改选项 expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
            create schema {self.schema_name} authorization
            {self.user_name}_noruser with blockchain;'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_sys '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertTrue(
            msg.count(self.Constant.CREATE_SCHEMA_SUCCESS_MSG) == 1,
            "执行失败:" + text)

        text = '----step5:普通用户登录数据库在防篡改schema下' \
               '创建防篡改用户表并插入数据 expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
            create table {self.schema_name}.{self.table_name}
            (id int primary key ,name varchar(20) NOT NULL ,
            address text DEFAULT 'xian');
            insert into {self.schema_name}.{self.table_name}
            values(1,'swf',DEFAULT),(2,'yyq','xiaomi');'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_noruser '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertIn(self.Constant.CREATE_TABLE_SUCCESS and
                      self.Constant.INSERT_SUCCESS_MSG, msg,
                      "执行失败:" + text)

        text = '----step6:系统管理员登录数据库使用copyto命令导出防篡改用户表' \
               ' expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''copy {self.schema_name}.{self.table_name} to 
                    '{self.user_path}';'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_sys '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertIn('COPY', msg, "执行失败:" + text)

        text = '----step7:初始用户用copyto命令导出防篡改历史表,全局区块表 ' \
               'expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''copy blockchain.{self.schema_name}_{self.table_name}_hist 
                    to '{self.hist_path}';
                    copy gs_global_chain to '{self.glob_path}';'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(sql_cmd)
        self.logger.info(msg)
        self.assertTrue(msg.count('COPY') == 2, "执行失败:" + text)

        text = '----step8:查询导出的表 expect:成功----'
        self.logger.info(text)
        cmd = f'''ls {macro.DB_INSTANCE_PATH} | grep 'user.bat';
                    ls {macro.DB_INSTANCE_PATH} | grep 'blockchain.hist';
                    ls {macro.DB_INSTANCE_PATH} | grep 'globaltable\''''
        self.logger.info(cmd)
        result = self.Common.get_sh_result(self.dbuserNode, cmd)
        self.assertIn('user.bat' and 'blockchain.hist' and 'globaltable',
                      result, "执行失败" + text)

        text = '----step9:普通用户登录数据库对防篡改表进行delete及select操作 ' \
               'expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
        delete from {self.schema_name}.{self.table_name} where name ='swf';
        delete from {self.schema_name}.{self.table_name} where name ='yyq';'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_noruser '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertTrue(msg.count(self.Constant.DELETE_SUCCESS_MSG) == 2,
                        "执行失败:" + text)

    def tearDown(self):
        text = '----step10:清理环境 expect:成功----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
            drop table  {self.schema_name}.{self.table_name} cascade;
            drop schema {self.schema_name} cascade;''')
        self.logger.info(sql_cmd)
        self.assertIn(self.Constant.DROP_SCHEMA_SUCCESS_MSG and
                      self.Constant.TABLE_DROP_SUCCESS, sql_cmd,
                      "执行失败:" + text)
        text = '----删除用户 expect:成功----'
        self.logger.info(text)
        drop_msg = []
        for user in self.user_privileges.keys():
            drop_cmd = f"drop user if exists {user} cascade; "
            msg = self.primary_sh.execut_db_sql(drop_cmd)
            self.logger.info(msg)
            drop_msg.append(msg)
        self.assertEqual(2,
                         drop_msg.count(self.Constant.DROP_ROLE_SUCCESS_MSG),
                         '执行失败:' + text)
        text = '-----恢复参数enable_copy_server_files默认值-----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
            alter system set enable_copy_server_files to {self.default_value};
            select pg_sleep(2);
            show enable_copy_server_files;''')
        self.logger.info(sql_cmd)
        self.assertEqual("off", sql_cmd.split("\n")[-2].strip(),
                         "执行失败:" + text)
        text = '-----删除文件-----'
        self.logger.info(text)
        del_file = f'''rm -rf {self.user_path};
                        rm -rf {self.hist_path};
                        rm -rf {self.glob_path}'''
        self.logger.info(del_file)
        self.Common.get_sh_result(self.dbuserNode, del_file)
        for s in [self.user_path, self.hist_path, self.glob_path ]:
            cmd = f'if [ -f {s} ]; then echo "does exists"; ' \
                f'else echo "not exists"; fi'
            file_judgments = self.Common.get_sh_result(self.dbuserNode, cmd)
            self.assertIn('not exists', file_judgments, "清理失败" + text)
        self.logger.info('--------------清理完成--------------')
        self.logger.info(f'-----{os.path.basename(__file__)} end-----')
